package com.lagou.client;

import com.lagou.zk.ServerInfoHandler;
import com.lagou.zk.ServerInfoVO;
import com.lagou.zk.ZkHelper;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ConcurrentHashMap;

/**
 * 连接管理器
 */
@Slf4j
public class ConnectionManager implements ServerInfoHandler {
    private static ConnectionManager connectionManager = new ConnectionManager();
    private final ConcurrentHashMap<String, RpcConsumer> consumerMap = new ConcurrentHashMap<>();

    public static ConnectionManager getManager() {
        return connectionManager;
    }

    /**
     * 监听ZK上的服务端状态变化
     *
     * @throws Exception
     */
    public void watchServer() throws Exception {
        ZkHelper.listen(this);
    }

    /**
     * 新服务器创建后，建立连接线程
     *
     * @param newServer
     * @throws InterruptedException
     */
    @Override
    public void onServerAdd(ServerInfoVO newServer) throws InterruptedException {
        synchronized (consumerMap) {
            boolean exists = consumerMap.get(newServer.getServerName()) != null;
            if (!exists && newServer.getIsUp()) {
                RpcConsumer consumer = RpcConsumer.create(newServer);
                new Thread(consumer).start();
                consumerMap.put(newServer.getServerName(), consumer);
                log.info("发现新服务[{}@{}:{}]", newServer.getServerName(), newServer.getIp(), newServer.getPort());
            } else {
                log.info(String.format("服务[%s]已存在", newServer.getServerName()));
            }
        }
    }

    /**
     * 服务器下线后，关闭连接线程
     *
     * @param rmvServer
     */
    @Override
    public void onServerRemove(ServerInfoVO rmvServer) {
        synchronized (consumerMap) {
            RpcConsumer consumer = consumerMap.get(rmvServer.getServerName());
            consumerMap.remove(rmvServer.getServerName());
            if (consumer != null) {
                consumer.shutdown();
                log.info("移除服务[{}@{}:{}]", rmvServer.getServerName(), rmvServer.getIp(), rmvServer.getPort());
            } else {
                log.info(String.format("服务[%s]不存在", rmvServer.getServerName()));
            }
        }
    }

    /**
     * 服务器状态变化时，增加或关闭连接线程
     *
     * @param server
     * @throws InterruptedException
     */
    @Override
    public void onServerUpdate(ServerInfoVO server) throws InterruptedException {
        if (server.getIsUp()) {
            onServerAdd(server);
        } else {
            onServerRemove(server);
        }
    }
}
